package com.shujia.spark.streaming

import org.apache.spark.sql.SparkSession
import org.apache.spark.streaming.dstream.DStream
import org.apache.spark.streaming.{Durations, StreamingContext}

object Demo07WindowOpt {
  def main(args: Array[String]): Unit = {

    // 窗口操作：在SparkStreaming只有基于时间的窗口，分为：滑动窗口和滚动窗口
    // 窗口：限定了每个计算批次的数据范围

    val spark: SparkSession = SparkSession
      .builder()
      .appName(this.getClass.getSimpleName.replace("$", ""))
      .master("local[2]") // 至少需要两个线程，因为接收数据需要一致占用一个线程
      .config("spark.default.parallelism", "2")
      .getOrCreate()

    // 构建StreamingContext：需要传入SparkContext以及Duration表示批次大小
    val ssc: StreamingContext = new StreamingContext(spark.sparkContext, Durations.seconds(5))

    val lineDS: DStream[String] = ssc.socketTextStream("master", 8888)

    lineDS
      .flatMap(_.split(","))
      .map(w => (w, 1))
      // 制定了一个滚动窗口，窗口的大小为5s，每隔5s会统计最近5s内的数据
      // 每个窗口间是独立的，没有交叠
      //      .reduceByKeyAndWindow(_ + _,Durations.seconds(5))
      // 滑动窗口，需要指定两个时间：滑动时间，窗口大小
      //      .reduceByKeyAndWindow((i1: Int, i2: Int) => i1 + i2, Durations.seconds(10), Durations.seconds(5))
      // 优化
      .reduceByKeyAndWindow((i1: Int, i2: Int) => i1 + i2, (i1: Int, i2: Int) => i1 - i2, Durations.seconds(10), Durations.seconds(5))
      // 每隔5s统计最近10s的数据
      .print()

    ssc.start()
    ssc.awaitTermination()

  }

}
